﻿using RdKafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Kafka.net451.Producer
{
    public class KafkaProducerUtil
    {
        public static void Producer()
        {
            string topicName = "V2X_perception";
            using (RdKafka.Producer producer = new RdKafka.Producer("10.102.1.31:9092"))
            using (Topic topic = producer.Topic(topicName))
            {
                Console.WriteLine("{" + producer.Name + "} producing on {" + topic.Name + "}. q to exit.");

                string text;
                while ((text = Console.ReadLine()) != "q")
                {
                    byte[] data = Encoding.UTF8.GetBytes(text);
                    Task<DeliveryReport> deliveryReport = topic.Produce(data);
                    var unused = deliveryReport.ContinueWith(task =>
                    {
                        Console.WriteLine("Partition: {" + task.Result.Partition + "}, Offset: {" + task.Result.Offset + "}");
                    });
                }
            }
        }

       
    }
}
